/*
 * Licensed to the Deppon EXPRESS .
 */
package com.depponframework.rocketmq.spring.core;

import com.depponframework.rocketmq.spring.support.RocketUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;

@Slf4j
public class RocketProducerWrapper {

    private final DefaultMQProducer producer;

    private String topic;
    private Integer sendMsgTimeout;

    public RocketProducerWrapper(String nameServerAddress, String producerGroup, String topic) {
        this(nameServerAddress, producerGroup, topic, 3000, 1024 * 1024 * 4);
    }

    public RocketProducerWrapper(String nameServerAddress, String producerGroup, String topic, Integer sendMsgTimeout, Integer maxMessageSize) {
        producer = new DefaultMQProducer();
        producer.setNamesrvAddr(nameServerAddress);
        producer.setProducerGroup(producerGroup);
        producer.setInstanceName(String.format("%s-%s-%s", this.getClass().getSimpleName(), topic, System.currentTimeMillis()));
        producer.setSendMsgTimeout(sendMsgTimeout);//default:3000
        producer.setMaxMessageSize(maxMessageSize);//default:1024 * 1024 * 4; it is 4M .

        this.topic = topic;
        this.sendMsgTimeout = sendMsgTimeout;
    }

    public DefaultMQProducer getProducer() {
        return producer;
    }

    //region producer's startup and shutdown
    public void startupProducer() throws MQClientException {
        producer.start();
        log.debug("default rocket mq producer for topic '{}' has been started!", this.topic);
    }

    public void shutdownProducer() {
        producer.shutdown();
        log.debug("default rocket mq producer for topic '{}' has been shutdown!", this.topic);
    }
    //endregion

    //region concurrently message
    //① async send
    public void sendAsync(Object value, String keys, String tags, SendCallback sendCallback) throws RemotingException, MQClientException, InterruptedException, IOException {
        Message msg = new Message(this.topic, tags, keys, RocketUtils.serialize(value));
        producer.send(msg, sendCallback, this.sendMsgTimeout);
    }

    //② sync send
    public SendResult sendSync(Object value, String keys, String tags) throws InterruptedException, RemotingException, MQClientException, MQBrokerException, IOException {
        Message msg = new Message(this.topic, tags, keys, RocketUtils.serialize(value));
        return producer.send(msg, this.sendMsgTimeout);
    }

    //③ sync batch send
    public SendResult sendSync(Collection<Object> values, String keys, String tags) throws InterruptedException, RemotingException, MQClientException, MQBrokerException, IOException {
        List<Message> msgList = new ArrayList<>();
        for (Object value : values) {
            msgList.add(new Message(this.topic, tags, keys, RocketUtils.serialize(value)));
        }
        return producer.send(msgList, this.sendMsgTimeout * values.size());
    }
    //endregion

    //region orderly message
    //endregion

    //region transaction message
    //endregion

}
